use embed_std::sync::arc::Arc;
use embed_std::thread::sleep_ms;
use embed_std::time::now;
use msgbus::bus;
use msgbus::capi::{get_bus, init_bus};
use msgbus::handler::MessageHandler;
use msgbus::infoln;
use msgbus::message::{CallReturn, Message, Packer, Unpacker};

struct Client;

impl MessageHandler for Client {
    fn handle(&self, msg: &Message) {
        infoln!(
            "notify: topic {}, content {}",
            msg.topic(),
            String::from_utf8_lossy(msg.content()).to_string()
        );
    }

    fn call(&self, topic: &str, method: &str, args: &[u8]) -> CallReturn {
        infoln!(
            "call: topic {} method {} args len {}",
            topic,
            method,
            args.len()
        );
        let mut unpacker = Unpacker::new(args);
        assert!(unpacker.fields() == 2);
        let seq = unpacker.get_u32();
        let s = String::from_utf8_lossy(unpacker.get_bytes());
        let mut ret: Vec<u8> = Vec::new();
        ret.extend_from_slice(b"call: ");
        ret.extend_from_slice(format!("#{} {}", seq, s).as_bytes());
        Ok(ret)
    }
}

fn main() {
    {
        let _a = Arc::new(10);
        let (msg, rx) = Message::new_call("aa", "bb", &[0]);
        drop(msg);
        drop(rx);
    }
    let client = Box::new(Client {});
    // init_bus(10000);
    let token = bus().subscribe("aa/bb", Some(client), None);
    let _ = get_bus();
    let _ = embed_std::thread::spawn(|| {
        for i in 0..10000 {
            bus().publish("aa/bb", format!("hellow world {}", i).as_bytes(), 1000);
            if i % 100 == 0 {
                sleep_ms(10);
            }
        }
    });
    sleep_ms(5000);

    let mut max = 0;
    let mut min = 0;
    let mut total = 0i64;
    let mut count = 0;
    let mut total_bytes = 0;
    for i in 0..1000 {
        // for i in 0..1 {
        let cur = now().as_secs();
        let mut packer = Packer::new();
        let content = format!("hello world {}", i);
        packer.put_u32(i as u32);
        packer.put_bytes(content.as_bytes());
        let data = packer.build();
        infoln!("call echo args {}", content);
        match bus().call("aa/bb", "echo", data.as_slice(), 1000) {
            Ok(data) => {
                infoln!("call return: {}", String::from_utf8_lossy(data.as_slice()));
                total_bytes += data.len();
                let delta = now().as_secs() - cur;
                if min == 0 || delta < min {
                    min = delta;
                }
                if delta > max {
                    max = delta;
                }
                total += delta as i64;
            }
            Err(e) => {
                infoln!("call failed: {}", e);
            }
        }
        count += 1;
        if count % 100 == 0 {
            sleep_ms(10);
        }
    }
    println!(
        "total {} us {} bytes count {} avg {} us max {} us min {} us",
        total,
        total_bytes,
        count,
        total / count as i64,
        max,
        min
    );

    sleep_ms(500000);
    bus().unsubscribe(token.as_str());
    loop {
        sleep_ms(1000);
    }
}
